package com.yhd.rabbit.deadletter;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;

import static com.yhd.rabbit.config.DeadLetterMqConfig.*;

/**
 * @author 二十
 * @since 2022/1/14 9:03 下午
 *
 *
 * spring.rabbitmq.host=localhost
 * spring.rabbitmq.port=5672
 * spring.rabbitmq.username=root
 * spring.rabbitmq.password=root
 * spring.rabbitmq.virtual-host=/yhd
 * #队列确认
 * spring.rabbitmq.publisher-confirm-type=correlated
 * spring.rabbitmq.publisher-returns=true
 * #默认情况下消息消费者是自动确认消息的，如果要手动确认消息则需要修改确认模式为manual
 * spring.rabbitmq.listener.simple.acknowledge-mode=manual
 * # 消费者每次从队列获取的消息数量。此属性当不设置时为：轮询分发，设置为1为：公平分发
 * spring.rabbitmq.listener.simple.prefetch=1
 * spring.rabbitmq.template.mandatory=true
 * spring.rabbitmq.listener.direct.acknowledge-mode=manual
 * spring.rabbitmq.listener.simple.default-requeue-rejected=false
 */
@Component
@Slf4j
public class DeadLetterMq {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 消息发送端
     *
     * @param msg
     */
    public void sendMsg(String msg) {
        rabbitTemplate.convertSendAndReceive(BUSINESS_EXCHANGE_NAME, "", msg);
    }

    /**
     * 业务接收端A
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = BUSINESS_QUEUEA_NAME)
    public void receiveA(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        //System.out.println("收到业务消息A："+ msg);
        boolean ack = true;
        Exception exception = null;
        try {
            if (msg.contains("RabbitMQ")) {
                throw new RuntimeException("dead letter exception");
            }
        } catch (Exception e) {
            ack = false;
            exception = e;
        }
        if (!ack) {
            log.error("消息消费发生异常，error msg:{}", exception.getMessage(), exception);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } else {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }


    //==================================//

    /**
     * 死信接收端A
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = DEAD_LETTER_QUEUEA_NAME)
    public void receiveA2(Message message, Channel channel) throws IOException {
        System.out.println("收到死信消息A：" + new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}
